#include "prmMQlib.h"

using namespace prmMQlib;
using namespace std;

#define	CONSUMER_MESSAGE_TYPE	"msgType"

////////////////////////////////////////////////////////////////////////////////
CPrmMQlib::CConsumer::CConsumer(
		Connection* mqConnection,
		const string& topicQueueName,
		ConsumerClient*	mesgCallback,
		bool isTopic,
		bool clientAck
	) : session (0),
		destination (0),
		consumer(0){
		this->useTopic = isTopic;
		this->clientAck = clientAck;
		this->topicQueueName = topicQueueName;
		this->mesgCallback = mesgCallback;
		this->mqConnection = mqConnection;
		this->isTemperery  = false;

		try {
			mqConnection->setExceptionListener(this);

			// Create a Session
			if( clientAck ) {
				session = mqConnection->createSession( Session::CLIENT_ACKNOWLEDGE );
			} else {
				session = mqConnection->createSession( Session::AUTO_ACKNOWLEDGE );
			}

			// Create the destination (Topic or Queue)
			if( useTopic ) {
				destination = session->createTopic( topicQueueName );
			} else {
				destination = session->createQueue( topicQueueName );
			}

			consumer = session->createConsumer( destination );
			consumer->setMessageListener( this );

		} catch (CMSException& e) {
			e.printStackTrace();
		}
}


CPrmMQlib::CConsumer::CConsumer(
		Connection* mqConnection,
		const string& topicQueueName,
		ConsumerClient*	mesgCallback,
		const string& userID
) {
		this->useTopic = true;
		this->clientAck = false;
		this->topicQueueName = topicQueueName;
		this->mesgCallback = mesgCallback;
		this->mqConnection = mqConnection;
		this->userID = userID;
		this->isTemperery  = false;

		try {
			mqConnection->setExceptionListener(this);

			// Create a Session
			session = mqConnection->createSession( Session::AUTO_ACKNOWLEDGE );
			string selector("msgType > 0");
			Topic* topic = session->createTopic( topicQueueName );
			consumer = session->createDurableConsumer(topic, userID, selector, false);
			consumer->setMessageListener( this );

		} catch (CMSException& e) {
			e.printStackTrace();
		}
}

CPrmMQlib::CConsumer::CConsumer(
		Connection* mqConnection,
		const string& topicQueueName,
		ConsumerClient*	mesgCallback,
		bool isTopic,
		bool clientAck,
		bool isTemp
) 	 : session (0),
		destination (0),
		consumer(0) {
		this->useTopic = isTopic;
		this->clientAck = clientAck;
		this->topicQueueName = topicQueueName;
		this->mesgCallback = mesgCallback;
		this->mqConnection = mqConnection;
		this->isTemperery  = isTemp;
		try {
			mqConnection->setExceptionListener(this);

			// Create a Session
			if( clientAck ) {
				session = mqConnection->createSession( Session::CLIENT_ACKNOWLEDGE );
			} else {
				session = mqConnection->createSession( Session::AUTO_ACKNOWLEDGE );
			}

			// Create the destination (Topic or Queue)
			if( useTopic ) {
				destination = session->createTopic( topicQueueName );
			} else {
				destination = session->createQueue( topicQueueName );
			}

			if( isTemperery ) {
	            replyProducer = session->createProducer(NULL);
	            replyProducer->setDeliveryMode(DeliveryMode::PERSISTENT);
			}

			consumer = session->createConsumer( destination );
			consumer->setMessageListener( this );

		} catch (CMSException& e) {
			e.printStackTrace();
		}
}


CPrmMQlib::CConsumer::~CConsumer() throw() {
	// Destroy resources.
	try{
		if( destination != NULL ) delete destination;
		cout << "destination destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	destination = NULL;

	try{
		if( consumer != NULL ) delete consumer;
		cout << "consumer destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	consumer = NULL;

	try{
		if( isTemperery == true && replyProducer ) delete replyProducer;
		cout << "replyProducer destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	replyProducer = NULL;

	try {
		if(mesgCallback != NULL) delete mesgCallback;
		cout << "mesgCallback destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }

	// Close open resources.
	try{
		if( session != NULL ) session->close();
		cout << "session close is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }

	try{
		if( session != NULL ) delete session;
		cout << "session destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	session = NULL;
}


void
CPrmMQlib::CConsumer::onMessage( const Message* message ) throw()
{
	if(message->propertyExists(CONSUMER_MESSAGE_TYPE) == false) {
		cout << "WARNING : Received wrong message" << endl;
		return;
	}

	if(message->getIntProperty(CONSUMER_MESSAGE_TYPE) == TXT_MESSAGE_TYPE ) {
		const TextMessage* textMessage = dynamic_cast< const TextMessage* >( message);
		const string txt = textMessage->getText();

		if(isTemperery == true) {
			mesgCallback->OnMessageRequest(TXT_MESSAGE_TYPE, (char*)txt.c_str(), (int)txt.length(), message->getCMSReplyTo(), message->getCMSCorrelationID());
		}
		else {
			mesgCallback->OnMessage(TXT_MESSAGE_TYPE, (char*)txt.c_str(), txt.length());
		}
	}
	else if(message->getIntProperty(CONSUMER_MESSAGE_TYPE) == BIN_MESSAGE_TYPE ) {
		const BytesMessage* bytesMessage = dynamic_cast< const BytesMessage*> (message);
		unsigned char* buff = bytesMessage->getBodyBytes();
		if(isTemperery == true) {
			mesgCallback->OnMessageRequest(BIN_MESSAGE_TYPE, (char*)buff, bytesMessage->getBodyLength(), message->getCMSReplyTo(), message->getCMSCorrelationID());
		}
		else {
			mesgCallback->OnMessage(BIN_MESSAGE_TYPE, (char*)buff, bytesMessage->getBodyLength()) ;
		}
	}
	else {
		cout << "WARNING : this is wrong message " << endl;
	}
}

void CPrmMQlib::CConsumer::onException( const CMSException& ex AMQCPP_UNUSED )
{
	mesgCallback->onException(ex);
	exit(1);
}

void
CPrmMQlib::CConsumer::transportInterrupted() {
	mesgCallback->transportInterrupted();
}

void
CPrmMQlib::CConsumer::transportResumed() {
	mesgCallback->transportResumed();
}


int
CPrmMQlib::CConsumer::replyMessage(
	const Destination* replyTo,
	const string messageId,
	string& strContent
){
	TextMessage* message = session->createTextMessage(strContent);
	message->setIntProperty(CONSUMER_MESSAGE_TYPE, TXT_MESSAGE_TYPE);
	message->setCMSCorrelationID(messageId);
	replyProducer->send(replyTo, message);
	delete message;
	return strContent.length();
}

int
CPrmMQlib::CConsumer::replyMessage(
		const Destination* replyTo,
		const string messageId,
		unsigned char* buffer,	/* Byte based message - it will be google proto buf message */
		int& len				/* the length of message sent 								*/
){
	BytesMessage* message = session->createBytesMessage(const_cast<const unsigned char*>(buffer),len);
	message->setIntProperty(CONSUMER_MESSAGE_TYPE, BIN_MESSAGE_TYPE);
	message->setCMSCorrelationID(messageId);
	replyProducer->send(replyTo, message);
	delete message;
	return len;
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CPrmMQlib::CProducer::CProducer(
			Connection* connection,
			const std::string& topicQueueName,
			bool isTopic,
			bool clientAck
) {
	this->mqConnection = connection;
	this->useTopic = isTopic;
	this->clientAck = clientAck;
	this->topicQueueName = topicQueueName;

	try {
	    // Create a Session
	    if( clientAck ) {
	        session = mqConnection->createSession( Session::CLIENT_ACKNOWLEDGE );
	    } else {
	    	session = mqConnection->createSession( Session::AUTO_ACKNOWLEDGE );
	    }

	    // Create the destination (Topic or Queue)
	    if( useTopic ) {
	        destination = session->createTopic( topicQueueName );
	    } else {
	        destination = session->createQueue( topicQueueName );
	    }

	    // Create a MessageProducer from the Session to the Topic or Queue
	    producer = session->createProducer( destination );
	    producer->setDeliveryMode( DeliveryMode::NON_PERSISTENT );
	} catch (CMSException& e) {
		e.printStackTrace();
	}
}

CPrmMQlib::CProducer::~CProducer() {

	// Destroy resources.
	try{
		if( destination != NULL ) delete destination;
		cout << "destination destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	destination = NULL;

	try{
		if( producer != NULL ) delete producer;
		cout << "producer destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	producer = NULL;

	// Close open resources.
	try{
		if( session != NULL ) {
			session->close();
			delete session;
		}
		cout << "session close is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }

	try{
		if( session != NULL ) delete session;
		cout << "session destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	session = NULL;
} ;


int
CPrmMQlib::CProducer::sendMessage(
	string& strContent
){
	TextMessage* message = session->createTextMessage(strContent);
	message->setIntProperty(CONSUMER_MESSAGE_TYPE, TXT_MESSAGE_TYPE);
	producer->send(message);
	delete message;
	return strContent.length();
}

int
CPrmMQlib::CProducer::sendMessage(
		unsigned char* buffer,	/* Byte based message - it will be google proto buf message */
		int& len				/* the length of message sent 								*/
){
	BytesMessage* message = session->createBytesMessage(const_cast<const unsigned char*>(buffer),len);
	message->setIntProperty(CONSUMER_MESSAGE_TYPE, BIN_MESSAGE_TYPE);
	producer->send(message);
	delete message;
	return len;
}


////////////////////////////////////////////////////////////////////////////////
CPrmMQlib::CPrmMQlib() {
	activemq::library::ActiveMQCPP::initializeLibrary();
}

CPrmMQlib::~CPrmMQlib(){

	producers.clear();
	consumers.clear();
	producer_topics.clear();
	consumer_topics.clear();
	activemq::library::ActiveMQCPP::shutdownLibrary();
}



PrmGeneralReturnCode
CPrmMQlib::initPrmMQlib(
		std::string& brokerURI
) {
	if(brokerURI.empty()) {
		std::cout << "this is null broker id" << std::endl;
		return PRMMQLIB_API_ERR_INVALID_PARAMS;
	}

	PrmGeneralReturnCode r =	PRMMQLIB_API_SUCCESS;
	this->brokerURI = brokerURI;

	try {
		auto_ptr<ActiveMQConnectionFactory> connectionFactory(
				new ActiveMQConnectionFactory( brokerURI ) );

        // Create a Connection
		mqConnection = connectionFactory->createConnection();
		mqConnection->start();

		std::cout << "mqConnection starts ... " << std::endl;

	}catch ( CMSException& e ) {
		e.printStackTrace();
	}


	return r;
}


/*	Initialization with clientID  */
PrmGeneralReturnCode
CPrmMQlib::initPrmMQlib(
		std::string& brokerURI,
		const string& clientID
)  {
	if(brokerURI.empty()) {
		std::cout << "this is null broker id" << std::endl;
		return PRMMQLIB_API_ERR_INVALID_PARAMS;
	}

	PrmGeneralReturnCode r =	PRMMQLIB_API_SUCCESS;
	this->brokerURI = brokerURI;

	try {
		auto_ptr<ActiveMQConnectionFactory> connectionFactory(
				new ActiveMQConnectionFactory( brokerURI ) );

        // Create a Connection
		mqConnection = connectionFactory->createConnection();
		mqConnection->setClientID(clientID);
		mqConnection->start();

		std::cout << "mqConnection starts ... " << std::endl;

	}catch ( CMSException& e ) {
		e.printStackTrace();
	}


	return r;
}

PrmGeneralReturnCode
CPrmMQlib::addProducer(
		const string& TopicOrQueueName,
		bool useTopic,
		bool clientAck,
		ConsumerClient* client
){
	if(TopicOrQueueName.empty() || consumers.containsKey(TopicOrQueueName) == true) {
		cout << "producer topic name [" << TopicOrQueueName.c_str() << "] exists consumer list" << endl;
		return PRMMQLIB_API_ERR_INVALID_PARAMS;
	}

	PrmGeneralReturnCode r = PRMMQLIB_API_ERR_UNKNOWN;

	if(producers.containsKey(TopicOrQueueName) == true ) {
		return PRMMQLIB_API_SUCCESS;
	}

	int re = 0;
    if(client) {
        CRequestProducer* requestProducer = new CRequestProducer(mqConnection,TopicOrQueueName,true,true, client);
        re = producers.putIfAbsent(TopicOrQueueName, requestProducer);
        if(! requestProducer || re == false ) return r;
    }
    else {
    	CProducer *node = new CProducer(mqConnection, TopicOrQueueName, useTopic, clientAck );
    	re = producers.putIfAbsent(TopicOrQueueName, node);
    	if(! node || re == false ) return r;
    }
    cout << "producer is inserted in the producers queue with [" << TopicOrQueueName.c_str() << "]" << endl;


	producer_topics.push_back(TopicOrQueueName);
    r = PRMMQLIB_API_SUCCESS;

	return r;
}


PrmGeneralReturnCode
CPrmMQlib::addProducerfromServiceId(
		const string& serviceId
){
	if(serviceId.empty() == true)
		return PRMMQLIB_API_ERR_INVALID_PARAMS;

	string TopicOrQueueName = getTopicNameFromServiceId(serviceId);
	if(producers.containsKey(TopicOrQueueName) == true) {
		cout << "producer topic name [" << TopicOrQueueName.c_str() << "] already in the producer list" << endl;
		return PRMMQLIB_API_SUCCESS;
	}

	PrmGeneralReturnCode r = PRMMQLIB_API_ERR_UNKNOWN;
	int re = 0;
	CProducer *node = new CProducer(mqConnection, TopicOrQueueName, true, false);
	re = producers.putIfAbsent(TopicOrQueueName, node);
	if(! node || re == false ) return r;
    cout << "producer is inserted in the producers queue with [" << TopicOrQueueName.c_str() << "]" << endl;


	producer_topics.push_back(TopicOrQueueName);
    r = PRMMQLIB_API_SUCCESS;

	return r;
}


string CPrmMQlib::getTopicNameFromServiceId(
		const string& serviceId
){
	vector<std::string> strings;
	istringstream f(serviceId);
	std::string s;
	while (std::getline(f, s, '.')) {
		strings.push_back(s);
	}

	stringstream mesg;
	mesg << "prm.order." << strings[3] << "." << strings[4]<< "." << strings[5];
	return (mesg.str());

}


PrmGeneralReturnCode
CPrmMQlib::addConsumer(
	ConsumerClient* client,
	const string& TopicOrQueueName,
	bool useTopic,
	bool clientAck
){
	if(TopicOrQueueName.empty() || producers.containsKey(TopicOrQueueName)) {
		return PRMMQLIB_API_ERR_INVALID_PARAMS;
	}

	PrmGeneralReturnCode r = PRMMQLIB_API_ERR_UNKNOWN;

	if(consumers.containsKey(TopicOrQueueName) == true ) {
		return PRMMQLIB_API_SUCCESS;
	}

    CConsumer *node = new CConsumer(mqConnection, TopicOrQueueName, client, useTopic, clientAck );
	consumers.putIfAbsent(TopicOrQueueName, node);
    cout << "consumers is inserted in the consumers queue with [" << TopicOrQueueName.c_str() << "]" << endl;

	consumer_topics.push_back(TopicOrQueueName);
    r = PRMMQLIB_API_SUCCESS;
    return r;
}

PrmGeneralReturnCode
CPrmMQlib::addConsumer(
	ConsumerClient* client,
	const string& TopicOrQueueName,
	bool useTopic,
	bool clientAck,
	bool isTemperery
){
	if(TopicOrQueueName.empty() || producers.containsKey(TopicOrQueueName)) {
		return PRMMQLIB_API_ERR_INVALID_PARAMS;
	}

	PrmGeneralReturnCode r = PRMMQLIB_API_ERR_UNKNOWN;

	if(consumers.containsKey(TopicOrQueueName) == true ) {
		return PRMMQLIB_API_SUCCESS;
	}

    CConsumer *node = new CConsumer(mqConnection, TopicOrQueueName, client, useTopic, clientAck, isTemperery );
	consumers.putIfAbsent(TopicOrQueueName, node);
    consumer_topics.push_back(TopicOrQueueName);
    cout << "consumers is inserted in the consumers queue with [" << TopicOrQueueName.c_str() << "]" << endl;
    r = PRMMQLIB_API_SUCCESS;
    return r;
}


PrmGeneralReturnCode
CPrmMQlib::addDurableConsumer(
	ConsumerClient* client,
	const string& TopicOrQueueName,
	const string& userID
){
	if(TopicOrQueueName.empty() || producers.containsKey(TopicOrQueueName)) {
		return PRMMQLIB_API_ERR_INVALID_PARAMS;
	}

	PrmGeneralReturnCode r = PRMMQLIB_API_ERR_UNKNOWN;
	if(consumers.containsKey(TopicOrQueueName) == true ) {
		return PRMMQLIB_API_SUCCESS;
	}

    CConsumer *node = new CConsumer(mqConnection, TopicOrQueueName, client, userID );
	consumers.putIfAbsent(TopicOrQueueName, node);
	consumer_topics.push_back(TopicOrQueueName);
    cout << "consumers is inserted in the consumers queue with [" << TopicOrQueueName.c_str() << "]" << endl;

    r = PRMMQLIB_API_SUCCESS;
    return r;
}


int
CPrmMQlib::sendTextMessage (
		const std::string& TopicOrQueueName,
		std::string& textMessage
)  {
	int len = 0;
	CProducer *node = producers.get(TopicOrQueueName);
	len = node->sendMessage(textMessage);

	return len;
}


int
CPrmMQlib::sendBytesMessage(
		const std::string& TopicOrQueueName,
		unsigned char* buffer,
		int& length,
		std::string correlation_mesg_id
){
	int len = 0;

	if(correlation_mesg_id.size() ==0 ) {
		CProducer *node = producers.get(TopicOrQueueName);
		if(node)
			len = node->sendMessage(buffer, length);
		else
			cout << "topic name of [" << TopicOrQueueName << "] is not correct or no linked to mqlib" << endl;
	}
	else {
		CRequestProducer *node = (CRequestProducer *)producers.get(TopicOrQueueName);
		if(node)
			len = node->sendMessage( correlation_mesg_id, buffer, length);
		else
			cout << "topic name of [" << TopicOrQueueName << "] is not correct or no linked to mqlib" << endl;
	}

	return len;
}





int
CPrmMQlib::replyTextMessage (
		const Destination* replyTo,
		const string message_id,
		const std::string& TopicOrQueueName,
		std::string& textMessage
)  {
	int len = 0;
	CConsumer *node = consumers.get(TopicOrQueueName);
	if(node->isTemperery == false)
		return 0;

	len = node->replyMessage(replyTo, message_id, textMessage);

	return len;
}


int
CPrmMQlib::replyBytesMessage(
		const Destination* replyTo,
		const string message_id,
		const std::string& TopicOrQueueName,
		unsigned char* buffer,
		int& length
){
	int len = 0;
	CConsumer *node = consumers.get(TopicOrQueueName);
	if(!node || node->isTemperery == false)
		return 0;

	len = node->replyMessage(replyTo, message_id, buffer, length);
	return len;
}


///////////////////////////////////////////////////////////////////
CPrmMQlib::CRequestProducer::CRequestProducer (
		Connection* connection,
		const std::string& topicName,
		bool isTopic,
		bool clientAck,
		ConsumerClient* msg_cb ): CProducer(connection, topicName, isTopic, clientAck)
{
	this->mesgCallback = msg_cb;
	tempDest = session->createTemporaryTopic();
    responseConsumer = session->createConsumer(tempDest);
    responseConsumer->setMessageListener(this);
}


void
CPrmMQlib::CRequestProducer::onMessage( const Message* message ) throw()
{
	if(message->propertyExists(CONSUMER_MESSAGE_TYPE) == false) {
		cout << "WARNING : Received wrong message" << endl;
		return;
	}

	if(message->getIntProperty(CONSUMER_MESSAGE_TYPE) == TXT_MESSAGE_TYPE ) {
		const TextMessage* textMessage = dynamic_cast< const TextMessage* >( message);
		const string txt = textMessage->getText();
		mesgCallback->OnMessage(TXT_MESSAGE_TYPE, (char*)txt.c_str(), txt.length());
	}
	else if(message->getIntProperty(CONSUMER_MESSAGE_TYPE) == BIN_MESSAGE_TYPE ) {
		const BytesMessage* bytesMessage = dynamic_cast< const BytesMessage*> (message);
		unsigned char* buff = bytesMessage->getBodyBytes();
		mesgCallback->OnMessage(BIN_MESSAGE_TYPE, (char*)buff, bytesMessage->getBodyLength()) ;
	}
	else {
		cout << "WARNING : this is wrong message" << endl;
	}
}


int
CPrmMQlib::CRequestProducer::sendMessage(
	const string messageId,
	string& strContent
){
	TextMessage* message = session->createTextMessage(strContent);
	message->setIntProperty(CONSUMER_MESSAGE_TYPE, TXT_MESSAGE_TYPE);
	message->setCMSReplyTo(tempDest);
	message->setCMSCorrelationID(messageId);
	producer->send(message);
	delete message;
	return strContent.length();
}



int
CPrmMQlib::CRequestProducer::sendMessage(
		const string messageId,
		unsigned char* buffer,	/* Byte based message - it will be google proto buf message */
		int& len				/* the length of message sent 								*/
){
	BytesMessage* message = session->createBytesMessage(const_cast<const unsigned char*>(buffer),len);
	message->setIntProperty(CONSUMER_MESSAGE_TYPE, BIN_MESSAGE_TYPE);
	message->setCMSCorrelationID(messageId);
	message->setCMSReplyTo(tempDest);
	producer->send(message);
	delete message;
	return len;
}


CPrmMQlib::CRequestProducer::~CRequestProducer() throw() {
	// Destroy resources
	try{
		if( tempDest != NULL ) delete tempDest;
		cout << "tempDest destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	tempDest = NULL;

	try{
		if( responseConsumer != NULL ) delete responseConsumer;
		cout << "responseConsumer destructor is called" << endl;
	}catch ( CMSException& e ) { e.printStackTrace(); }
	responseConsumer = NULL;

}

